package cn.tannn.sink;

import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class MysqlSink extends RichSinkFunction<String>{

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    Connection connection;
    PreparedStatement iStmt,dStmt,uStmt;
    private Connection getConnection() {
        Connection conn = null;
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            String url = "jdbc:mysql://192.168.0.51:3316/test?serverTimezone=Asia/Shanghai&useSSL=false";
            conn = DriverManager.getConnection(url,"root","root");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return conn;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String insertSql = "insert into base_trademark(`id`, `tm_name`, `logo_url`) values (?,?,?)";
        String deleteSql = "delete from base_trademark where id=?";
        String updateSql = "update base_trademark set tm_name=?,logo_url=? where id=?";
        iStmt = connection.prepareStatement(insertSql);
        dStmt = connection.prepareStatement(deleteSql);
        uStmt = connection.prepareStatement(updateSql);
    }

    // 每条记录插入时调用一次
    @Override
    public void invoke(String value, Context context) throws Exception {
        Gson t = new Gson();
        JsonObject hs = t.fromJson(value, JsonObject.class);
        JsonObject source = hs.get("source").getAsJsonObject();
        String database = source.get("db").getAsString();
        String table = source.get("table").getAsString();
        // 操作类型
        String type =  hs.get("op").getAsString();


        if ("flink".equals(database) && "base_trademark".equals(table)) {
            if (OperationTypeEnum.INSERT.getAbbr().equals(type)) {
                logger.info("insert => " + value);
                JsonObject data = hs.get("after").getAsJsonObject();
                String id = verifyNull(data.get("id"));
                String tmName = verifyNull(data.get("tm_name"));
                String logoUrl = verifyNull(data.get("logo_url"));
                iStmt.setString(1, id);
                iStmt.setString(2, tmName);
                iStmt.setString(3, logoUrl);
                iStmt.executeUpdate();
            }else if (OperationTypeEnum.DELETE.getAbbr().equals(type)) {
                logger.info("delete => " + value);
                JsonObject data = hs.get("before").getAsJsonObject();
                String id = verifyNull(data.get("id"));
                dStmt.setString(1, id);
                dStmt.executeUpdate();
            }else if (OperationTypeEnum.UPDATE.getAbbr().equals(type)) {
                logger.info("update => " + value);
                JsonObject data = hs.get("after").getAsJsonObject();
                String id = verifyNull(data.get("id"));
                String tmName = verifyNull(data.get("tm_name"));
                String logoUrl = verifyNull(data.get("logo_url"));
                uStmt.setString(1, tmName);
                uStmt.setString(2, logoUrl);
                uStmt.setString(3, id);
                int count = uStmt.executeUpdate();
                if(count<=0){
                        //不存在需要被更新的数据，那就插入
                    iStmt.setString(1, id);
                    iStmt.setString(2, tmName);
                    iStmt.setString(3, logoUrl);
                    iStmt.executeUpdate();
                }
            }else {
                logger.info("read => " + value);
            }

        }
    }
    @Override
    public void close() throws Exception {
        super.close();

        if(iStmt != null) {
            iStmt.close();
        }
        if(dStmt != null) {
            dStmt.close();
        }
        if(uStmt != null) {
            uStmt.close();
        }

        if(connection != null) {
            connection.close();
        }
    }


    private static String verifyNull(JsonElement jsonElement) {
        if(jsonElement.isJsonNull()){
            return null;
        }
        return jsonElement.getAsString();
    }
}
